位移主题 _consumer_offsets

Kafka 社区自 0.8.2.x 版本 开始在新版本 Consumer 中正式推出了全新的位移管理 机制，自然也包括这个新的位移主题。

新版本 Consumer 的位移管理机制其实也很简单，就是将 Consumer 的位移数据作为一 条条普通的 Kafka 消息，提交到 __consumer_offsets 中。可以这么说， __consumer_offsets 的主要作用是保存 Kafka 消费者的位移信息。它要求这个提交过程 不仅要实现高持久性，还要支持高频的写操作。显然，Kafka 的主题设计天然就满足这两个 条件。

位移主题就是普通的 Kafka 主题。你 可以手动地创建它、修改它，甚至是删除它。只不过，它同时也是一个内部主题，大部分情 况下，你其实并不需要“搭理”它，也不用花心思去管理它，把它丢给 Kafka 就完事了。

虽说位移主题是一个普通的 Kafka 主题，但它的消息格式却是 Kafka 自己定义的，用户不 能修改，也就是说你不能随意地向这个主题写消息，因为一旦你写入的消息不满足 Kafka 规定的格式，那么 Kafka 内部无法成功解析，就会造成 Broker 的崩溃。事实上，Kafka Consumer 有 API 帮你提交位移，也就是向位移主题写消息。千万不要自己写个 Producer 随意向该主题发送消息。

消息格式

首先从 Key 说起。一个 Kafka 集群中的 Consumer 数量会有很多，既然这个主题保存的是 Consumer 的位移数据，那么消息格式中必须要有字段来标识这个位移数据是哪个 Consumer 的。这种数据放在哪个字段比较合适呢？显然放在 Key 中比较合适。

位移主题的 Key 中应该保存 3 部分内容：。

接下来，我们再来看看消息体的设计。虽然看起来保存一个位移值就可以了。实际上，社区的方案要复杂得多，比如消息体还保存了位移提交的一些其他元数 据，诸如时间戳和用户自定义的数据等。保存这些元数据是为了帮助 Kafka 执行各种各样 后续的操作，比如删除过期位移消息等。但总体来说，还是可以简单地认为消息体就是 保存了位移值。

位移主题如何创建

当 Kafka 集群中的第一个 Consumer 程序启动时，Kafka 会自动创建位移主题。但如果是 Kafka自动创建的，分区数是怎么设置的呢？这就要看 Broker 端参数 offsets.topic.num.partitions 的 取值了。它的默认值是 50，因此 Kafka 会自动创建一个 50 分区的位移主题。 Kafka 日志路径下冒出很多 __consumer_offsets-xxx 这样的目录，就是 Kafka 自动帮你创建的位移主题。

副本数或备份因子是 Broker 端另一个参数 offsets.topic.replication.factor 要做的事情了。它的默认值是 3。

提交位移

目前有自动提交，手动提交两种。

Consumer 端有个参数叫 enable.auto.commit，如果值是 true，则 Consumer 在后台默 默地为你定期提交位移，提交间隔由一个专属的参数 auto.commit.interval.ms 来控制。 自动提交位移有一个显著的优点，就是省事，但这一点同时也是缺点。那就是丧失了很大的灵活性和可控性，完全没法把控 Consumer 端的位移管理。

手动提交位移，即设置 enable.auto.commit = false。 一旦设置了 false，作为 Consumer 应用开发的你就要承担起位移提交的责任。Kafka Consumer API 为你提供了位移提交的方法，如 consumer.commitSync 等。当调用这些 方法时，Kafka 会向位移主题写入相应的消息。

如果你选择的是自动提交位移，那么就可能存在一个问题：只要 Consumer 一直启动着，它就会无限期地向位移主题写入消息。

我们来举个极端一点的例子。假设 Consumer 当前消费到了某个主题的最新一条消息，位移是 100，之后该主题没有任何新消息产生，故 Consumer 无消息可消费了，所以位移永远保持在 100。由于是自动提交位移，位移主题中会不停地写入位移 =100 的消息。显然 Kafka 只需要保留这类消息中的最新一条就可以了，之前的消息都是可以删除的。这就要求 Kafka 必须要有针对位移主题消息特点的消息删除策略，否则这种消息会越来越多，最终撑爆整个磁盘。

Kafka 是怎么删除位移主题中的过期消息的呢？答案就是 Compaction。国内很多文献都将其翻译成压缩。

Kafka 使用Compact 策略来删除位移主题中的过期消息，避免该主题无限期膨胀。那么应该如何定义 Compact 策略中的过期呢？对于同一个 Key 的两条消息 M1 和 M2，如果 M1 的发送时间早于 M2，那么 M1 就是过期消息。Compact 的过程就是扫描日志的所有消息，剔除那些过期的消息，然后把剩下的消息整理在一起。

Kafka 提供了专门的后台线程定期地巡检待 Compact 的主题，看看是否存在满足条件的可删除数据。这个后台线程叫 Log Cleaner。很多实际生产环境中都出现过位移主题无限膨胀占用过多磁盘空间的问题，如果你的环境中也有这个问题，建议你去检查一下 Log Cleaner 线程的状态，通常都是这个线程挂掉了导致的。